package com.ibm.factories;

import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

import org.apache.log4j.Logger;

import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Throwables;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.publisher.LogPublisher;
import com.ibm.cps.publisher.MessageQueuePublisher;
import com.ibm.cps.publisher.MongoPublisher;
import com.ibm.cps.publisher.SeyrenRuleMongoPublisher;
import com.ibm.interfaces.AbstractMessagePublisher;
import com.ibm.interfaces.IPersistency;
import com.ibm.mongo.MongoPersistency;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

public class PublishProcessorFactory {

	private static Map<String, Class<? extends AbstractMessagePublisher>> type2Publisher = new HashMap<String, Class<? extends AbstractMessagePublisher>>();
	private static Logger logger = Logger.getLogger(PublishProcessorFactory.class);
	private Map<String, AbstractMessagePublisher> name2Publisher;
	
	private static PublishProcessorFactory _instance = null;
	public PublishProcessorFactory() {
		name2Publisher = new HashMap<String, AbstractMessagePublisher>();
	}
	/*
	public synchronized static PublishProcessorFactory getInstance() throws CPSException{
		if(_instance == null){
			_instance = new PublishProcessorFactory();
		}
		return _instance;
	}

	private PublishProcessorFactory() throws CPSException {
		name2Publisher = new HashMap<String, AbstractMessagePublisher>();
		loadExistedPublisher();
	}
	*/
	static {
		type2Publisher.put(MessageFields.MESSAGEQUEUE, MessageQueuePublisher.class);
		type2Publisher.put(MessageFields.LOG_TYPE, LogPublisher.class);
		type2Publisher.put(MessageFields.SEYRENRULE, SeyrenRuleMongoPublisher.class);
		type2Publisher.put(MessageFields.MONGO_PUBLISHER, MongoPublisher.class);
	}

	public static void addNewPublishProcessor(String type, Class<? extends AbstractMessagePublisher> clzz) {
		type2Publisher.put(type, clzz);
	}

	public static AbstractMessagePublisher createOutputProcessor(String metadata) throws CPSException {
		ObjectNode node = JSONUtility.fromObjectString(metadata);
		JSONUtility.checkNotNull(node, MessageFields.NAME);
		if (node.get(MessageFields.DESTINATION) == null) {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, "Null publisher destination.");
		} else if (type2Publisher.get(node.get(MessageFields.DESTINATION).asText()) == null) {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR,
					"Unkown publisher destination " + node.get(MessageFields.DESTINATION).asText());
		}

		String destination = node.get(MessageFields.DESTINATION).asText();
		try {
			Constructor<? extends AbstractMessagePublisher> constructor = type2Publisher.get(destination)
					.getConstructor(String.class);
			return constructor.newInstance(metadata);
		} catch (InvocationTargetException e) {
			Throwable exception = e.getTargetException();
			if (exception instanceof CPSException) {
				throw (CPSException) exception;
			}
			logger.error(Throwables.getStackTraceAsString(e));
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "Fail to create publisher: " + e.getCause());
		} catch (Throwable other) {
			logger.error(Throwables.getStackTraceAsString(other));
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "Internal error.");
		}
	}

	public AbstractMessagePublisher addPublishProcessor(String metadata) throws CPSException {
		AbstractMessagePublisher publisher = createOutputProcessor(metadata);
		if (name2Publisher.containsKey(publisher.getName())) {
			throw new CPSException(ErrorCode.ALREADY_EXIST_CODE,
					"Publish processor " + publisher.getName() + " already existed.");
		}
		name2Publisher.put(publisher.getName(), publisher);
		return publisher;
	}

	public AbstractMessagePublisher deletePublishProcessor(String name, String tenantid) throws CPSException {
		if (name2Publisher.containsKey(name) && name2Publisher.get(name).getTenantid().equals(tenantid)) {
			AbstractMessagePublisher removedPublisher = name2Publisher.get(name);
			name2Publisher.remove(name);
			logger.info("Publisher processor " + name + " is deleted.");
			return removedPublisher;
		} else {
			throw new CPSException(ErrorCode.INVALID_REQUEST_CODE,
					"Publish processor " + name + " for tenant " + tenantid + " doesn't exist.");
		}
	}

	public ObjectNode getOutputProcessor(String name, String tenantid) throws CPSException {
		if (!(name2Publisher.containsKey(name) && name2Publisher.get(name).getTenantid().equals(tenantid))) {
			throw new CPSException(ErrorCode.INVALID_REQUEST_CODE,
					"Publish processor " + name + " for tenant " + tenantid + " doesn't exist.");
		}
		AbstractMessagePublisher publisher = name2Publisher.get(name);
		ObjectNode node = JSONUtility.fromObjectString(publisher.getProcessorMetadta());
		return node;
	}

	public boolean containsPublisher(String publisher) {
		return name2Publisher.containsKey(publisher);
	}

	private void loadExistedPublisher() {
		try {
			IPersistency persistency = MongoPersistency.getInstace();
			Collection<String> publishers = persistency.getPublisher(null, null);
			if (publishers != null) {
				for (String publisher : publishers) {
					try {
						addPublishProcessor(publisher);
					} catch (CPSException e) {
						logger.warn("Error publihser metadata stored in mongo:" + publisher);
					}
				}
			}
		} catch (CPSException e) {
			logger.error(Throwables.getStackTraceAsString(e));
		}
	}

	public static boolean isPublishProcessor(String type) {
		return type.equals(MessageFields.PUBLISHER_TYPE);
	}
}
